package hn.cch.flink;

import hn.cch.flink.demo.Tokenizer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class FlinkKafka {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String bootstrapServers = "192.168.10.153:9092";
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", bootstrapServers);
        properties.setProperty("transaction.timeout.ms", "60000");
        String topic = "test";

        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers(bootstrapServers)
            .setTopics(topic)
            .setGroupId("flink-kafka")
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        DataStreamSource<String> dataStreamSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");


        // dataStreamSource.print();
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = dataStreamSource.flatMap((new Tokenizer()))
            .keyBy((KeySelector<Tuple2<String, Integer>, Object>) tuple2 -> tuple2.f0).sum(1);
        sum.print();

        // KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
        //     .setBootstrapServers(bootstrapServers)
        //     .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        //         .setTopic(topic)
        //         .setValueSerializationSchema(new SimpleStringSchema())
        //         .build()
        //     )
        //     .setKafkaProducerConfig(properties)
        //     .setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        //     .build();
        // DataStreamSink<String> dataStreamSink = dataStreamSource.sinkTo(kafkaSink);

        env.execute("jobName");
    }

}
